package kafka

import (
	"logAgent/utils"
	"time"

	"github.com/Shopify/sarama"
)

var (
	client      sarama.SyncProducer //kafka客户端对象
	logDataChan chan *logData       //日志信息管道
)

//logData 日志信息
type logData struct {
	topic string //主题
	data  string //数据
}

//Init 初始化kafka连接
func Init(addr string, size int) error {
	config := sarama.NewConfig()                              //新建配置
	config.Producer.RequiredAcks = sarama.WaitForAll          //设置ack=all,要求leader和follower全部执行完毕才行
	config.Producer.Partitioner = sarama.NewRandomPartitioner //新建一个随机分区
	config.Producer.Return.Successes = true                   //成功交付的消息将在success channel中返回
	//连接kafka
	var err error
	client, err = sarama.NewSyncProducer([]string{addr}, config)
	if err != nil {
		utils.Log.Error("初始化kafka失败: ", err)
		return err
	}
	//初始化管道
	logDataChan = make(chan *logData, size)
	//发送信息
	go sendToKafka()
	return nil
}

//SendToChan 将信息发送到管道
func SendToChan(topic, value string) {
	logData := logData{
		topic: topic,
		data:  value,
	}
	logDataChan <- &logData
}

//sendToKafka 发送信息到kafka
func sendToKafka() {
	for {
		select {
		case message := <-logDataChan:
			//构造一个消息
			msg := &sarama.ProducerMessage{}
			msg.Topic = message.topic                      //设置主题
			msg.Value = sarama.StringEncoder(message.data) //添加内容
			//发送消息
			_, _, err := client.SendMessage(msg)
			if err != nil {
				utils.Log.Error("发送日志信息到kafka失败")
				return
			}
			utils.Log.Error("发送信息到kafka了")
		default:
			time.Sleep(time.Millisecond * 100)
		}
	}
}
